package cc.verywell.pureblock.msg.nfs.idx;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.AbstractMap;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.function.BiConsumer;

import cc.verywell.pureblock.func.node.KeyShell;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.msg.nfs.NfsAutoStoreRule;
import cc.verywell.pureblock.msg.nfs.NfsControler;
import cc.verywell.pureblock.msg.nfs.NfsLocalStoreProvider;
import cc.verywell.pureblock.msg.nfs.NfsPolicyV1;
import cc.verywell.pureblock.msg.nfs.hd.LfsHdMainV1;
import cc.verywell.pureblock.msg.nfs.hd.LfsHdSupply;
import cc.verywell.pureblock.msg.nfs.hd.NoEnoughSpaceException;
import cc.verywell.pureblock.msg.nfs.idx.extraDownload.DownloadManager;
import cc.verywell.pureblock.msg.nfs.idx.extraDownload.DownloadManager.DownloadCallback;
import cc.verywell.pureblock.msg.nfs.idx.extraDownload.DownloadNeed;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadManager;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadManager.UploadCallback;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadNeed;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadSupply;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadingMonitor;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.DataLink;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.Sign;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.Sub;
import cc.verywell.pureblock.net.AioProvider;
import cc.verywell.pureblock.net.NormalSession;
import cc.verywell.pureblock.util.sign.Sha;

/**
* 
* 程序功能有2：a做种 b下载
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月5日 下午1:22:30
*/
public class NfsIdxCloudMap {
	public long REMOTE_TIMEOUT = 0;
	public long NEWSTORE_DELAY = 200;
	
	final private NfsidxMain idxmain;
	final private LfsHdSupply localHd; 
	final UploadManager extraUp;
	final DownloadManager extraDown;
	final private Executor exec;
	final private NfsIdxCloudCache mem;
	private NfsControler nfs;
	public NfsIdxCloudMap(Executor exec, Path localPath, long hdCapacity, NfsIdxCloudCache mem) {
		NfsidxMainV0 im = new NfsidxMainV0(uploadSupply);
		this.idxmain = im;
		this.localHd = new LfsHdMainV1(localPath, im, hdCapacity);
		this.extraUp = new UploadManager(uploadNeed);
		this.extraDown = new DownloadManager(downloadNeed);
		this.exec = exec;
		//nfs加载已分离到下方;
		this.mem = mem;
	}
	public void bindNfs(NfsPolicyV1 nfs) {
		synchronized (unstartRemote) {
			this.nfs = nfs;
			nfs.setUploadListenerIn(uploadSupply);
			///自动缓存策略留到最后才启动;
			nfs.setNfsAutoStock(nfsAutoStock);
			exec.execute(()->startRemoteRun());
		}
	}
	
	//// 上传返回
	public UploadCallback upload(Path filePath, List<Sub> alias, List<Sign> signs, List<DataLink> extralinks) {
		///做种选项未配置
		return extraUp.upload(filePath, alias, signs, extralinks);
	}
	//// 下载返回
	public DownloadCallback download(Path logPath) {
		return extraDown.download(logPath);
	}
	public DownloadCallback download(final KeyShell mainKey, Path logPath, String targetPath) {
		return extraDown.download(mainKey, logPath, targetPath);
	}
	
	
	
	private NfsAutoStoreRule nfsAutoStock = new NfsAutoStoreRule() { //store合并分发组件
		@Override
		public void newStored(KeyShell... fileHash) {
			//Map<KeyShell, Set<KeyShell>> map = new HashMap<>(fileHash.length);
			//for (KeyShell key : fileHash) map.put(key, null);
			//nfs.newStored(map);
			for (KeyShell key : fileHash) {
				newStoredCollect(key, null);
			}
		}
		@Override
		public void fireNewFileStored(KeyShell key, byte[] payload, Set<KeyShell> editAbleFrom, boolean isHandled) {
			if(isHandled)return; //如果是主动下载的，不用分发文件, 不用内部保存;
			
			boolean ok = (mem!=null && mem.contain(key));
			if(!ok) {
				try {
					localHd.putTemp(key, payload);
					ok = true;
				} catch (IOException e) {
					if(Pblog.FS)Pblog.FS(e);
				}
				ok = (mem!=null && mem.put(key, payload)) || ok;
				
			}
			if(ok)newStoredCollect(key,editAbleFrom);
		}
	};
	
	private UploadSupply uploadSupply = new UploadSupply(){
		@Override
		public void fireStoreFile(NormalSession nc, List<byte[]> fileHash) {
			extraUp.fireStoreFile(nc, fileHash);
		}
		@Override
		public void fireWantFile(NormalSession nc, byte[] fileHash) {
			extraUp.fireWantFile(nc, fileHash);
		}
		@Override
		public NfsidxStructure needNewExternalFile(KeyShell key, NfsidxStructure errFile) {
			return extraUp.needNewExternalFile(key, errFile);
		}
	};

	private UploadNeed uploadNeed = new UploadNeed() {
		@Override
		public byte addIndex(KeyShell key, NfsidxStructure idx) {
			byte result = idxmain.addIndex(key, idx);
			if(result==NfsidxMain.ADD_OK) {
				if(nfs!=null) {
					BiConsumer<KeyShell, byte[]> b = nfs.fireLocalFileAdded(key);
					if(b!=null) {
						byte[] bs = getLocalData(key);
						if(bs!=null)b.accept(key, bs);
					}
				}
				newStoredCollect(key, null); //需要在此函数内异步推送newStored(key)
			}
			return result;
		}
		@Override
		public byte replaceIndex(KeyShell key, NfsidxStructure oldidx, NfsidxStructure newidx) {
			return idxmain.replaceIndex(key, oldidx, newidx);
		}
		@Override
		public byte removeIndex(KeyShell key, NfsidxStructure idx) {
			return idxmain.removeIndex(key, idx);
		}
		@Override
		public void removeHold(Object lock) {
			localHd.removeHold(lock);
		}
		@Override
		public NfsidxStructure putAndHoldInternal(Object hold, KeyShell key, byte[] value) throws IOException,NoEnoughSpaceException {
			if(mem!=null)mem.put(key,value);
			NfsidxStructure result = localHd.putAndHoldInternal(hold, key, value);
			//稍后会在upload调用addIndex时一起推送//newStoredCollect(key, null); //需要在此函数内异步推送newStored(key)
			return result;
		}
		@Override
		public NfsidxStructure tryHoldInternal(Object hold, KeyShell key, byte[] value) {
			if(mem!=null)mem.put(key,value);
			try {
				return localHd.putAndHoldInternal(hold, key, value);
			} catch (IOException e) {
				if(Pblog.FS)Pblog.FS("tryHoldInternal [",hold.hashCode(),"] -",LfsHdSupply.formatSize(value.length)," ",e);
				return null;
			}
		}
	};
	
	private DownloadNeed downloadNeed = new DownloadNeed() {
		
		@Override
		public UploadingMonitor getUploadMonitor(String absolutePath) {
			return extraUp.getUploadMonitor(absolutePath);
		}
		
		@Override
		public void getRemoteData(BiConsumer<KeyShell, byte[]> func, KeyShell... key) {
			if(nfs!=null)nfs.newSeeking(func, REMOTE_TIMEOUT, key);
			else {
				LinkedList<Map.Entry<BiConsumer<KeyShell, byte[]>, KeyShell[]>> s = unstartRemote;
				synchronized (s) {
					if(nfs!=null)nfs.newSeeking(func, REMOTE_TIMEOUT, key);
					else s.add(new AbstractMap.SimpleEntry<>(func,key));
				}
				
			}
		}
		
		@Override
		public byte[] getLocalData(KeyShell key) {
			return NfsIdxCloudMap.this.getLocalData(key);
		}
		
		@Override
		public void cancelRemoteDownload(BiConsumer<KeyShell, byte[]> func, KeyShell... key) {
			// TODO 现阶段对外接口NfsPolicyV1无此功能;
		}
	};
	
	final public NfsLocalStoreProvider fsmap = new NfsLocalStoreProvider() {
		@Override
		public byte[] get(KeyShell key) {
			return getLocalData(key);
		}
		
		@Override
		public boolean contain(KeyShell key) {
			return (mem!=null && mem.contain(key)) || idxmain.getIndex(key)!=null;
		}
	}; 
	
	public KeyShell putLocalData(byte[] value) {
		KeyShell key = new KeyShell(Sha.getInstance().hash(value));
		boolean ok = (mem!=null && mem.contain(key)); ///1.加入cache
		if(!ok) {
			try {
				localHd.putTemp(key, value);///2.加入硬盘
				ok = true;
			} catch (IOException e) {
				if(Pblog.FS)Pblog.FS(e);
			}
			ok = (mem!=null && mem.put(key, value)) || ok;
		}
		if(ok) {
			if(nfs!=null) {///3.看看是否在等
				BiConsumer<KeyShell, byte[]> b = nfs.fireLocalFileAdded(key);
				if(b!=null)b.accept(key, value);
			}
			newStoredCollect(key,null);///4.发布到网络
			return key;
		}
		return null;
	}
	public byte[] getLocalData(KeyShell key) {
		if(mem!=null) {
			byte[] bs = mem.get(key);
			if(bs!=null)return bs;
			bs = getLocalHDData(key);
			if(bs==null)return null;
			mem.put(key,bs);
			return bs;
		}else
		return getLocalHDData(key);
	}
	
	private byte[] getLocalHDData(KeyShell key) {
		NfsidxStructure idx = idxmain.getIndex(key);
		while(idx!=null) {
			NfsidxStructure oldidx = idx;
			try {
				byte[] bs;
				if(idx.isExternal()) {
					bs = readExtra(idx);
					//if(bs.length>AioProvider.FILE_MAX_SIZE)System.err.println("extra read err:"+idx.getSize()+" return "+bs.length+" limit "+AioProvider.FILE_MAX_SIZE);
				}else {
					bs = localHd.read(idx);
					//if(bs.length>AioProvider.FILE_MAX_SIZE)System.err.println("localHd err:"+idx.getSize()+" return "+bs.length+" limit "+AioProvider.FILE_MAX_SIZE);
				}
				return bs;
			} catch (IOException e) {
				if(Pblog.FS)Pblog.FS(e);
				idx = uploadSupply.needNewExternalFile(key, oldidx);
				if(idx==null)return null;
				if(oldidx.equals(idx))return null;
				idxmain.replaceIndex(key, oldidx, idx);
			}
		}
		return null;
	}
	
	static public byte[] readExtra(NfsidxStructure idx) throws IOException {
		if(idx.isExternal() && idx.isPuredata()) {
			Path p = Paths.get(idx.getPath());
			long off,size;
			if(idx.isPartfile()) {
				off = idx.getOffset();
				size = idx.getSize()+1;
			}else {
				off = 0;
				size = idx.getSize()+1;//Files.size(p)+1;
			}
			if(size<=0 || size>AioProvider.FILE_MAX_SIZE) throw new IllegalArgumentException("readExtra size too big:"+String.valueOf(idx));
			try( SeekableByteChannel channel = Files.newByteChannel(Paths.get(idx.getPath()),StandardOpenOption.READ).position(off); ) {
				byte[] bs = new byte[(int)size];
				bs[0]=0;
				ByteBuffer bb = ByteBuffer.wrap(bs, 1, bs.length-1);
				channel.read(bb);
				if(bb.hasRemaining())
					throw new IOException("readExtra file size err: "+bb.capacity()+" => "+bb.position()+" at "+idx.getPath()+":"+off);
				return bs;
			}
		}else {
			throw new IllegalArgumentException("readExtra Idx need isExternal Ture and  isPuredata Ture -"+String.valueOf(idx));
		}
	}
	
	private LinkedList<Map.Entry<BiConsumer<KeyShell, byte[]>, KeyShell[]>> unstartRemote = new LinkedList<>();
	private void startRemoteRun() {
		LinkedList<Map.Entry<BiConsumer<KeyShell, byte[]>, KeyShell[]>> s = unstartRemote;
		boolean start = false;
		synchronized (s) {
			if(nfs!=null) {
				start = true;
				unstartRemote = null;
				for (Entry<BiConsumer<KeyShell, byte[]>, KeyShell[]> entry : s) {
					nfs.newSeeking(entry.getKey(), REMOTE_TIMEOUT, entry.getValue());
				}
			}
		}
		if(start==true && newStoredLock.tryAcquire()) {
			collectRun();
		}
	}
	
	private Semaphore newStoredLock = new Semaphore(1);
	private BlockingQueue<Entry<KeyShell,Set<KeyShell>>> newStoredQueue = new LinkedBlockingQueue<>();
	private void newStoredCollect(KeyShell key,Set<KeyShell> editAbleFrom) {
		try {
			newStoredQueue.put(new AbstractMap.SimpleImmutableEntry<>(key,editAbleFrom));
		} catch (InterruptedException e) {
			if(Pblog.IO)Pblog.FS("NfsAutoStoreRule err: ",e);
			if(nfs!=null)nfs.newStored(Collections.singletonMap(key, editAbleFrom));
			return;
		}
		if(nfs!=null && newStoredLock.tryAcquire()) {
			try {
				exec.execute(()->collectRun());
			} catch (Exception e) {
				newStoredLock.release();
				throw e;
			}
		}
	}
	///store分发器需要运行的内容;
	private void collectRun() {
		newStoredLock.release();
		do{
			if(newStoredLock.tryAcquire()) {
				try {
					try {
						Thread.sleep(NEWSTORE_DELAY);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					collectRunCore();
				}finally {
					newStoredLock.release();
				}
			}else {
				break;
			}
		}while(null!=newStoredQueue.peek());
	}
	private void collectRunCore() {
		if(newStoredQueue.isEmpty())return;
		Map<KeyShell, Set<KeyShell>> map = new HashMap<>(newStoredQueue.size());
		Entry<KeyShell,Set<KeyShell>> entry;
		while(null!=(entry=newStoredQueue.poll())){
			///putIfAbsent 会比较 value是否为null;
			///根据测试 LangTest#Map_putIfAbsent_isit_for_null制定
			Set<KeyShell> oldset = map.putIfAbsent(entry.getKey(), entry.getValue());
			if(oldset!=null && entry.getValue()!=null) {
				oldset.addAll(entry.getValue());
			}
//			Set<KeyShell> oldset = map.put(entry.getKey(), entry.getValue());
//			if(oldset!=null) {
//				if(entry.getValue()==null)map.put(entry.getKey(), oldset);
//				else entry.getValue().addAll(oldset);
//			}
		}
		if(!map.isEmpty())
			nfs.newStored(map); //此处会缓慢;
	}
	
}
